package com.klm.easymq.core.impl;

import com.klm.easymq.core.MonitorService;
import lombok.extern.slf4j.Slf4j;

import java.time.LocalDateTime;
import java.time.format.DateTimeFormatter;
import java.util.*;
import java.util.concurrent.ConcurrentHashMap;

/**
 * 内存监控服务实现
 * 使用内存存储监控数据，适用于测试环境
 */
@Slf4j
public class MemoryMonitorService implements MonitorService {
    
    private final Map<String, Map<String, Object>> messageRecords = new ConcurrentHashMap<>();
    private final Map<String, Map<String, Object>> topicStats = new ConcurrentHashMap<>();
    private final DateTimeFormatter formatter = DateTimeFormatter.ofPattern("yyyy-MM-dd HH:mm:ss");
    
    @Override
    public void recordMessageSend(String traceId, String topic, String message, String status) {
        Map<String, Object> record = new HashMap<>();
        record.put("traceId", traceId);
        record.put("topic", topic);
        record.put("message", message);
        record.put("status", status);
        record.put("type", "SEND");
        record.put("timestamp", LocalDateTime.now().format(formatter));
        
        messageRecords.put(traceId, record);
        updateTopicStats(topic, status);
        
        log.debug("Recorded message send: traceId={}, topic={}, status={}", traceId, topic, status);
    }
    
    @Override
    public void recordMessageConsume(String traceId, String topic, String status, String errorMsg) {
        Map<String, Object> record = messageRecords.get(traceId);
        if (record != null) {
            record.put("consumeStatus", status);
            record.put("consumeTime", LocalDateTime.now().format(formatter));
            if (errorMsg != null) {
                record.put("errorMsg", errorMsg);
            }
        } else {
            // 如果没有发送记录，创建新的消费记录
            record = new HashMap<>();
            record.put("traceId", traceId);
            record.put("topic", topic);
            record.put("status", status);
            record.put("type", "CONSUME");
            record.put("timestamp", LocalDateTime.now().format(formatter));
            if (errorMsg != null) {
                record.put("errorMsg", errorMsg);
            }
            messageRecords.put(traceId, record);
        }
        
        updateTopicStats(topic, status);
        log.debug("Recorded message consume: traceId={}, topic={}, status={}", traceId, topic, status);
    }

    @Override
    public void recordConsumerLifecycle(String consumerId, String group, String topic, String eventType, String details) {

    }

    @Override
    public void recordConnectionStatus(String adapterType, String status, String errorMsg) {

    }

    @Override
    public void recordRetryEvent(String traceId, String topic, int attempt, String errorMsg) {

    }

    @Override
    public void recordIdempotentSkip(String traceId, String topic, String reason) {

    }

    @Override
    public List<Map<String, Object>> queryByTraceId(String traceId) {
        Map<String, Object> record = messageRecords.get(traceId);
        if (record != null) {
            return Arrays.asList(record);
        }
        return Collections.emptyList();
    }

    @Override
    public List<Map<String, Object>> queryFailedMessages(String topic, String startTime, String endTime, int limit) {
        return List.of();
    }

    @Override
    public List<Map<String, Object>> queryConsumerLifecycle(String consumerId, String group, String startTime, String endTime) {
        return List.of();
    }

    @Override
    public List<Map<String, Object>> queryConnectionHistory(String adapterType, String startTime, String endTime) {
        return List.of();
    }

    @Override
    public Map<String, Object> getTopicStats(String topic) {
        return topicStats.getOrDefault(topic, new HashMap<>());
    }

    @Override
    public Map<String, Object> getFailureStats(String startTime, String endTime) {
        return Map.of();
    }

    @Override
    public void cleanupExpiredData(int retentionDays) {
        // 内存存储不需要清理，数据会随着应用重启而清空
        log.info("Memory monitor service cleanup called, but no cleanup needed for memory storage");
    }
    
    @Override
    public String getType() {
        return "memory";
    }
    
    /**
     * 更新主题统计信息
     */
    private void updateTopicStats(String topic, String status) {
        Map<String, Object> stats = topicStats.computeIfAbsent(topic, k -> new HashMap<>());
        
        // 更新发送统计
        if ("SUCCESS".equals(status)) {
            stats.put("sendSuccess", ((Integer) stats.getOrDefault("sendSuccess", 0)) + 1);
        } else if ("FAILED".equals(status)) {
            stats.put("sendFailed", ((Integer) stats.getOrDefault("sendFailed", 0)) + 1);
        }
        
        // 更新消费统计
        if ("SUCCESS".equals(status)) {
            stats.put("consumeSuccess", ((Integer) stats.getOrDefault("consumeSuccess", 0)) + 1);
        } else if ("FAILED".equals(status)) {
            stats.put("consumeFailed", ((Integer) stats.getOrDefault("consumeFailed", 0)) + 1);
        }
        
        stats.put("lastUpdate", LocalDateTime.now().format(formatter));
    }
} 